package org.sws.infrastructure.event;

import com.alibaba.fastjson.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.sws.infrastructure.persistent.dao.ITaskDao;
import org.sws.infrastructure.persistent.po.Task;
import org.sws.types.enums.TaskState;
import org.sws.types.event.BaseEvent;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author sws
 * @Date 2025/3/20 09:35
 * @description: 消息发送
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class EventPublisher {

    private final RabbitTemplate rabbitTemplate;
    private final ThreadPoolExecutor executor;
    private final ITaskDao taskDao;

    /**
     * 发布消息
     * @param topic 消息主题
     * @param eventMessage 消息体
     */
    public void publish(String topic, BaseEvent.EventMessage<?> eventMessage) {
        String jsonMessage = JSON.toJSONString(eventMessage);
        publish(topic,jsonMessage);
    }

    /**
     * 发布消息
     * @param topic 消息主题
     * @param jsonEventMessage 消息体
     */
    public void publish(String topic, String jsonEventMessage) {
        try {
            rabbitTemplate.convertAndSend(topic, jsonEventMessage);
            log.info("发送MQ消息 topic:{} message:{}", topic, jsonEventMessage);
        } catch (AmqpException e) {
            log.error("发送MQ消息失败 topic:{} message:{}", topic, JSON.toJSONString(jsonEventMessage), e);
            throw e;
        }
    }

    public void syncPublish(Task task) {
        executor.execute(() ->{
            try {
                rabbitTemplate.convertAndSend(task.getTopic(),task.getMessage());
                // 更新任务表 记为完成
                task.setState(TaskState.complete.getCode());
                taskDao.updateTaskSendStatus(task);
            }catch (Exception e){
                log.error("异步发送消息事件异常,topic:{}, task:{}",task.getTopic(),task);
                // 更新任务表，记为失败
                task.setState(TaskState.fail.getCode());
                taskDao.updateTaskSendStatus(task);
            }
        });
    }

    public void syncPublish(List<Task> taskList) {
        executor.execute(() ->{
            for (Task task : taskList){
                try {
                    rabbitTemplate.convertAndSend(task.getTopic(),task.getMessage());
                    // 更新任务表 记为完成
                    task.setState(TaskState.complete.getCode());
                    taskDao.updateTaskSendStatus(task);
                }catch (Exception e){
                    log.error("异步发送消息事件异常,topic:{}, task:{}",task.getTopic(),task);
                    // 更新任务表，记为失败
                    task.setState(TaskState.fail.getCode());
                    taskDao.updateTaskSendStatus(task);
                }
            }
        });
    }
}
